package com.qingyunge.app.dws;

import com.alibaba.fastjson.JSONObject;
import com.qingyunge.bean.SkuEvalInfoFinly;
import com.qingyunge.bean.YearPriceLocationClassBean;
import com.qingyunge.util.MyClickHouseUtil;
import com.qingyunge.util.MyKafkaUtil;
import com.qingyunge.util.MyfenPriceUtil;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple4;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.util.Collector;

import java.time.Duration;

public class Dws_year_price_location_window {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment().setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        String topic = "sku_eval_topic";
        String gourpid = "Dws_year_price_location_skuClass";
        DataStreamSource<String> skuEvalDS = env.addSource(MyKafkaUtil.getFlinkKafkaConsumer(topic, gourpid));
        SingleOutputStreamOperator<YearPriceLocationClassBean> flatMapDS = skuEvalDS.flatMap(new FlatMapFunction<String, YearPriceLocationClassBean>() {
            @Override
            public void flatMap(String s, Collector<YearPriceLocationClassBean> collector) throws Exception {
                try {
                    SkuEvalInfoFinly skuEvalInfoFinly = JSONObject.parseObject(s, SkuEvalInfoFinly.class);
                    int year;
                    String quarter;
                    try{
                        int month = Integer.parseInt(skuEvalInfoFinly.getDate().substring(5, 7));
                        quarter = MyfenPriceUtil.getQuarter(month);
                    }catch (Exception e){
                        quarter = MyfenPriceUtil.getQuarter(1);
                    }
                    try{
                        year = Integer.parseInt(skuEvalInfoFinly.getDate().substring(0, 4));
                    }catch (Exception e){
                        year = 2020;
                    }
                    Long pingLunNum = skuEvalInfoFinly.getEvalNum();
                    Double price = skuEvalInfoFinly.getPrice();
                    String salaryRange = MyfenPriceUtil.getSalaryRange(price);
                    int rangeIndex = MyfenPriceUtil.getRangeIndex(price);
                    String skuClass = skuEvalInfoFinly.getSkuId().substring(0, 1);
                    String location = skuEvalInfoFinly.getLocation();
                    Long ts = skuEvalInfoFinly.getTs();
                    YearPriceLocationClassBean priceLocationClassBean = new YearPriceLocationClassBean(year,
                          quarter,rangeIndex , salaryRange, location, skuClass, 1L, price, pingLunNum, ts);
                    collector.collect(priceLocationClassBean);
                } catch (Exception e) {
                    e.printStackTrace();
                    System.out.println("s = " + s);
                }
            }
        });
        SingleOutputStreamOperator<YearPriceLocationClassBean> yearPriceLocationWaterMark = flatMapDS.assignTimestampsAndWatermarks(WatermarkStrategy.<YearPriceLocationClassBean>forBoundedOutOfOrderness(Duration.ofSeconds(2)).withTimestampAssigner(new SerializableTimestampAssigner<YearPriceLocationClassBean>() {
            @Override
            public long extractTimestamp(YearPriceLocationClassBean yearPriceLocationClassBean, long l) {
                return yearPriceLocationClassBean.getTs();
            }
        }));

        // TODO 分组 开窗
        SingleOutputStreamOperator<YearPriceLocationClassBean> reduceDS = yearPriceLocationWaterMark.keyBy(new KeySelector<YearPriceLocationClassBean, Tuple4<String, String, String, String>>() {
            @Override
            public Tuple4<String, String, String, String> getKey(YearPriceLocationClassBean value) throws Exception {
                return Tuple4.of(value.getQuarter(), value.getPriceQj(), value.getLocation(), value.getSkuClass());
            }
        }).window(TumblingEventTimeWindows.of(Time.seconds(10)))
                .reduce(new ReduceFunction<YearPriceLocationClassBean>() {
                    @Override
                    public YearPriceLocationClassBean reduce(YearPriceLocationClassBean yearPriceLocationClassBean, YearPriceLocationClassBean t1) throws Exception {
                        yearPriceLocationClassBean.setNum(yearPriceLocationClassBean.getNum() + t1.getNum());
                        yearPriceLocationClassBean.setPingLunNum(yearPriceLocationClassBean.getPingLunNum() + t1.getPingLunNum());
                        yearPriceLocationClassBean.setPriceSum(yearPriceLocationClassBean.getPriceSum() + t1.getPriceSum());
                        return yearPriceLocationClassBean;
                    }
                });
        reduceDS.print("=======>");
        reduceDS.addSink(MyClickHouseUtil.getSinkFunction("" +
                "insert into dws_year_priceQj_location_sku_class_table " +
                "values (?,?,?,?,?,?,?,?,?,?)"));
        env.execute();
    }
}
